package org.halk.job_client.mq.cosumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.halk.job_client.component.JobContext;
import org.halk.job_client.mq.produce.JobProduce;
import org.halk.job_client.pojo.entity.JobRequest;
import org.halk.job_client.pojo.entity.JobResponse;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @Author haozhu7
 * @Date 2024/9/27 23:25
 **/
@Slf4j
@Component
@RocketMQMessageListener(topic = JobConsumer.TOPIC_JOB, consumerGroup = JobConsumer.TOPIC_JOB, messageModel = MessageModel.CLUSTERING)
public class JobConsumer  implements RocketMQListener<JobRequest> {

    public static final String TOPIC_JOB = "HALK_JOB";

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private JobProduce jobProduce;

    private static String LOCK_KEY = "JOB:KEY:";

    @Override
    public void onMessage(JobRequest jobRequest) {
        log.info("job_consumer receive msg: {}", jobRequest);
        try {
            //需要加锁，防止并发执行
            Boolean b = stringRedisTemplate.opsForValue().setIfAbsent(LOCK_KEY + jobRequest.getScheduleId(), "1", jobRequest.getRedisTimeOut(), TimeUnit.SECONDS);
            if (b == null || !b) {
                //发送拒绝执行响应
                jobProduce.sendJobMsg(new JobResponse().setMsgId(jobRequest.getMsgId()).setScheduleId(jobRequest.getScheduleId()).setStatus(0));
                log.info("job_consumer receive msg，任务锁未释放: {}，", jobRequest);
                return;
            }
            JobContext.run(jobRequest);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
            stringRedisTemplate.delete(LOCK_KEY + jobRequest.getScheduleId());
        }
    }

}
